package abacab.remote;

import static abacab.ext.RemotingExtension.DEFAULT_PORT;
import static abacab.ext.RemotingExtension.PROTOCOL;

import java.io.IOException;
import java.net.ConnectException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import abacab.ActorSystem;
import abacab.ext.SerializationExtension;
import abacab.ext.SerializationExtensionProvider;
import abacab.ref.ActorRef;
import abacab.remote.message.FindActor;
import abacab.remote.message.Message;
import abacab.remote.protocol.Acceptor;
import abacab.remote.protocol.ImplementationFactory;
import abacab.remote.ref.RemoteActorRef;
import abacab.remote.utils.RemoteUtils;

import com.typesafe.config.ConfigException.BadValue;

/**
 * Registers all the connections with remote actor systems
 */
@SuppressWarnings("rawtypes")
public final class ConnectionRegistry {
	/** The logger */
	private static final Logger																	logger						= LoggerFactory
																																										.getLogger(ConnectionRegistry.class);

	/** The factory to use to create the implementations for remoting */
	private static final String																	REMOTE_FACTORY		= "abacab.remote.factory";

	/** The timeout to get an actor */
	private static final long																		TIMEOUT_GET_ACTOR	= 30000L;

	/** The known remote actors */
	private static final ConcurrentMap<String, RemoteActorRef>	remoteActors			= new ConcurrentHashMap<String, RemoteActorRef>();

	/** The actor system */
	private final ActorSystem																		actorSystem;

	/** The network interface */
	private final InetAddress																		netInterface;

	/** The listening port */
	private final int																						listeningPort;

	/** The implementation used for remoting */
	private final ImplementationFactory													factory;

	/** The extension for serilization */
	private final SerializationExtension												serilizationExt;

	/** The connections */
	private final ConcurrentMap<ConnectionKey, Connection>			connections;

	/** The threads waiting for a response */
	private final ConcurrentMap<String, Object>									waiter;

	/** The acceptor */
	private Acceptor																						acceptor;

	/** Indicates that the registry is stopped */
	private volatile boolean																		stopped;

	/**
	 * Create the registry
	 * 
	 * @param actorSystem
	 *          the actor system
	 * @param netInterface
	 *          the network interface
	 * @param port
	 *          the listening port
	 */
	@SuppressWarnings("unchecked")
	public ConnectionRegistry(ActorSystem actorSystem, InetAddress netInterface, int port) {
		this.actorSystem = actorSystem;
		this.netInterface = netInterface;
		this.listeningPort = port;
		String remoteFactory = this.actorSystem.getConfig().getString(REMOTE_FACTORY);
		Class<ImplementationFactory> factoryClass;
		try {
			factoryClass = (Class<ImplementationFactory>) Class.forName(remoteFactory);
		}
		catch (ClassNotFoundException e) {
			throw new BadValue(REMOTE_FACTORY, "The remote implementation's factory (" + remoteFactory + ") does not exist",
					e);
		}
		catch (ClassCastException e) {
			throw new BadValue(REMOTE_FACTORY, "The remote implementation's factory (" + remoteFactory
					+ ") does not implement \"" + ImplementationFactory.class.getName() + "\"");
		}
		try {
			this.factory = factoryClass.newInstance();
		}
		catch (Exception e) {
			throw new BadValue(REMOTE_FACTORY,
					"The remote implementation's factory (" + remoteFactory + ") can not be built", e);
		}
		this.serilizationExt = actorSystem.getExtension(SerializationExtensionProvider.INSTANCE);
		this.connections = new ConcurrentHashMap<ConnectionKey, Connection>();
		this.waiter = new ConcurrentHashMap<String, Object>();
	}

	/**
	 * Initialize the acceptor
	 * 
	 * @throws IOException
	 *           if the acceptor can not be created
	 */
	public void init() throws IOException {
		this.acceptor = this.factory.buildAcceptor(this);
	}

	/**
	 * Stop the connections
	 */
	public void stop() {
		this.stopped = true;
		this.acceptor.stop();
	}

	/**
	 * @return the actor system
	 */
	public ActorSystem getActorSystem() {
		return this.actorSystem;
	}

	/**
	 * @return the network interface
	 */
	public InetAddress getNetworkInterface() {
		return this.netInterface;
	}

	/**
	 * @return the listening port
	 */
	public int getListeningPort() {
		return this.listeningPort;
	}

	/**
	 * @return the extension for serilization
	 */
	public SerializationExtension getSerilizationExt() {
		return this.serilizationExt;
	}

	/**
	 * Register a new connection
	 * 
	 * @param remoteAddress
	 *          the remote address
	 * @param connection
	 *          the connection
	 * @return the old connection (if any)
	 * @throws IllegalAccessException
	 *           if the remoting is stopped
	 */
	public Connection registerConnection(InetSocketAddress remoteAddress, Connection connection)
			throws IllegalAccessException {
		if (this.stopped) {
			throw new IllegalAccessException("Remoting is stopped");
		}
		return this.connections.put(
				new ConnectionKey(remoteAddress.getAddress().getHostAddress(), remoteAddress.getPort()), connection);
	}

	/**
	 * Unregister a connection
	 * 
	 * @param remoteAddress
	 *          the remote address
	 * @param connection
	 *          the connection
	 */
	public void unregisterConnection(InetSocketAddress remoteAddress, Connection connection) {
		this.connections.remove(new ConnectionKey(remoteAddress.getAddress().getHostAddress(), remoteAddress.getPort()),
				connection);
	}

	/**
	 * Get a connection
	 * 
	 * @param hostname
	 *          the host's name
	 * @param port
	 *          the port
	 * @return a connection to the host
	 * @throws ConnectException
	 *           if the connection could not be made
	 */
	public Connection getConnection(String hostname, int port) throws ConnectException {
		if (this.stopped) {
			throw new ConnectException("Remoting is stopped");
		}
		Connection cnx = this.connections.get(new ConnectionKey(hostname, port));
		if (cnx == null) {
			if (this.stopped) {
				throw new ConnectException("Remoting is stopped");
			}
			cnx = this.factory.buildClient(this, hostname, port);
			// No need to add it to the map since it will be done once the connection is established
		}
		return cnx;
	}

	/**
	 * Get the "remote" view of a local actor
	 * 
	 * @param actorRef
	 *          the local actor's reference
	 * @return the full path (as seen by a remote actor system) for this actor
	 */
	public <T> String getRemotePath(ActorRef<T> actorRef) {
		if (actorRef == null) {
			return null;
		}
		return PROTOCOL + getNetworkInterface().getHostAddress() + ":" + getListeningPort() + actorRef.getPath();
	}

	/**
	 * Get a waiter
	 * 
	 * @param remotePath
	 *          the remote path
	 * @return the waiter
	 */
	private Object getWaiter(String remotePath) {
		Object obj = new Object();
		Object concObj = this.waiter.putIfAbsent(remotePath, obj);
		if (concObj != null) {
			obj = concObj;
		}
		return obj;
	}

	/**
	 * Remove a waiter
	 * 
	 * @param remotePath
	 *          the remote path
	 */
	private void notifyWaiter(String remotePath) {
		Object obj = this.waiter.remove(remotePath);
		if (obj != null) {
			synchronized (obj) {
				obj.notifyAll();
			}
		}
	}

	/**
	 * Get the reference to a remote actor
	 * 
	 * @param remotePath
	 *          the path to the remote actor
	 * @return the reference
	 * @throws IllegalArgumentException
	 *           if the path is invalid
	 * @throws IOException
	 *           if the connection or the sending failed
	 * @throws InterruptedException
	 *           if the thread was interrupted while waiting for the actor
	 */
	public RemoteActorRef getRemoteActor(String remotePath) throws IOException, IllegalArgumentException,
			InterruptedException {
		if (remotePath == null) {
			return null;
		}
		RemoteActorRef ref = remoteActors.get(remotePath);
		if (ref == null) {
			Connection connection = getConnection(remotePath);
			if (connection == null) {
				return null;
			}
			Object monitor = getWaiter(remotePath);
			synchronized (monitor) {
				ref = remoteActors.get(remotePath);
				if (ref != null) {
					return ref;
				}
				connection.send(new Message(null, null, new FindActor(remotePath, null)));
				monitor.wait(TIMEOUT_GET_ACTOR);
			}
			ref = remoteActors.get(remotePath);
		}
		return ref;
	}

	/**
	 * Get the reference to a remote actor
	 * 
	 * @param remotePath
	 *          the path to the remote actor
	 * @param cnx
	 *          the connection
	 * @return the reference
	 * @throws IllegalArgumentException
	 *           if the path is invalid
	 */
	public RemoteActorRef getRemoteActor(String remotePath, Connection cnx) throws IllegalArgumentException {
		if (remotePath == null) {
			return null;
		}
		RemoteActorRef ref = remoteActors.get(remotePath);
		if (ref == null) {
			if (cnx == null) {
				try {
					cnx = getConnection(remotePath);
				}
				catch (ConnectException e) {
					logger.error("Failed to get the connection for remote actor \"" + remotePath + "\"", e);
				}
			}
			ref = new RemoteActorRef(remotePath, cnx);
			RemoteActorRef concRef = remoteActors.putIfAbsent(remotePath, ref);
			if (concRef != null) {
				ref = concRef;
			}
		}
		return ref;
	}

	/**
	 * Get a connection to the remote system hosting the requested actor
	 * 
	 * @param remotePath
	 *          the remote path
	 * @return the connection
	 * @throws IllegalArgumentException
	 *           if the path is invalid
	 * @throws ConnectException
	 *           if the connection could not be made
	 */
	public Connection getConnection(String remotePath) throws IllegalArgumentException, ConnectException {
		if (!remotePath.startsWith(PROTOCOL)) {
			// Not a remote path
			return null;
		}
		int idx = remotePath.indexOf('/', PROTOCOL.length());
		if (idx < 0) {
			throw new IllegalArgumentException("A remote path must provider a host & optionaly a port: " + remotePath);
		}
		int idx2 = remotePath.indexOf(':', PROTOCOL.length());
		String remoteHost;
		int port;
		if (idx2 > 0 && idx - idx2 > 1) {
			remoteHost = remotePath.substring(PROTOCOL.length(), idx2);
			try {
				port = Integer.parseInt(remotePath.substring(idx2 + 1, idx));
			}
			catch (NumberFormatException e) {
				throw new IllegalArgumentException(
						"Wrong port number for remote actor: " + remotePath.substring(idx2 + 1, idx), e);
			}
		}
		else {
			remoteHost = remotePath.substring(PROTOCOL.length(), idx);
			port = DEFAULT_PORT;
		}
		return getConnection(remoteHost, port);
	}

	/**
	 * Handle received messages
	 * 
	 * @param cnx
	 *          the connection
	 * @param message
	 *          the received message
	 */
	@SuppressWarnings("unchecked")
	public void messageReceived(Connection cnx, Message message) {
		ActorRef ref = message.getRecipient();
		if (ref == null) {
			handleAdministrativeMessage(cnx, message);
		}
		else {
			ref.send(message.getMessage(), message.getSender());
		}
	}

	/**
	 * Handle an administrative message
	 * 
	 * @param cnx
	 *          the connection
	 * @param message
	 *          the message
	 */
	public void handleAdministrativeMessage(Connection cnx, Message message) {
		Object msg = message.getMessage();
		if (msg instanceof FindActor) {
			FindActor fa = (FindActor) msg;
			if (fa.getResult() == null) {
				ActorRef ref = ActorSystem.findActor(RemoteUtils.toLocalPath(fa.getActorPath()));
				try {
					cnx.send(new Message(null, null, new FindActor(fa.getActorPath(), Boolean.valueOf(ref != null))));
				}
				catch (IOException e) {
					// TODO: handle the error correctly
					logger.error("Failed to send a message", e);
				}
			}
			else if (fa.getResult().booleanValue()) {
				getRemoteActor(fa.getActorPath(), cnx);
				notifyWaiter(fa.getActorPath());
			}
			else {
				notifyWaiter(fa.getActorPath());
				logger.warn("Remote actor {} does not exist", fa.getActorPath());
			}
		}
		else {
			logger.error("Unknown type of message: {}", message.getClass().getName());
		}
	}
}